package org.gl.rtsp;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.net.MalformedURLException;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.nio.CharBuffer;
import java.util.HashSet;

import org.gl.nio.CameraProxyConnectPool;

/**
 * Refactor<br>
 * proxy RTSP request
 * ----------------<br>
 * @time 14-11-17 针对NAT后的摄像头代理无法被代理服务器主动连接，特此增加反向代理机制和NIO长连接服务，以支持内网网关主动上线  ，反向连接内网的摄像头代理
 * ----------------<br>
 * @author GeLiang 14-08-12
 */
public class RTSPProxy {
	private static final int BUFFER_SIZE = 8192;
	private static final int PROXY_PORT = 1800;
	private static final String CRLF = "\r\n";
	private static final String charset = "UTF-8";
	/**
	 * #私有协议规则,规定了RTSP代理中所需要的必要信息</br> rtsp://%s:%s?c=rtsp://%s:%s&w=%s</br>
	 * 代理服务器地址和端口+目标RTSP服务地址和端口+内网代理识别码
	 */
	private static final String PROXYURLFORMAT = "rtsp://%s:%s?c=rtsp://%s:%s&w=%s";

	/**
	 * 生成一个私有协议规则的转发链接，用于协商转发流程</br> see:{@link #PROXYURLFORMAT}
	 * 
	 * @param proxyHost
	 * @param proxyPort
	 * @param targetHost
	 * @param targetPort
	 * @param uuid
	 *            for gateway Serial Number
	 * @return
	 */
	public static String genProxyURL(String proxyHost, int proxyPort,
			String targetHost, int targetPort, String uuid) {
		return String.format(PROXYURLFORMAT, proxyHost, proxyPort, targetHost,
				targetPort, uuid);
	}

	/**
	 * proxy server url rtsp://192.168.4.45:<br>
	 * target server url rtsp://192.168.2.64:5540
	 */
	public static final String URL = "rtsp://192.168.4.45:" + PROXY_PORT
			+ "/proxy?url=rtsp://192.168.2.64:5540/";
	private final int TIMEOUT_SOCKET = 3000;
	CameraProxyConnectPool mCameraProxyConnectPool;

	private void start() throws IOException {
		ServerSocket serversocket = new ServerSocket(PROXY_PORT);
		mCameraProxyConnectPool = CameraProxyConnectPool.getInstance();
		while (true) {
			new WorkerThread(serversocket.accept()).start();
		}
	}

	class WorkerThread extends Thread implements Runnable {
		private final Socket mClientSocket;
		private Socket targetSocket;
		/**
		 * the BufferedOutputStream of client
		 */
		private final BufferedOutputStream mOutput;
		/**
		 * the BufferedReader of client
		 */
		private final BufferedReader mInput;
		final CharBuffer receiveBuf = CharBuffer.allocate(BUFFER_SIZE);
		private Request request;
		boolean isInitProxyInfo;

		boolean isClosedSocket;
		/**
		 * the BufferedInputStream of proxy target
		 */
		private BufferedInputStream s_mInput;
		/**
		 * the BufferedWriter of proxy target
		 */
		private BufferedWriter s_mOutput;

		// Each client has an associated session

		public WorkerThread(final Socket client) throws IOException {
			log.i(tag, "======================================");
			log.i(tag, "============              ============");
			log.i(tag, "============              ============");
			log.i(tag, "Client:" + client.getInetAddress().getHostAddress()
					+ ":" + client.getPort());
			client.setSoTimeout(TIMEOUT_SOCKET);
			mInput = new BufferedReader(new InputStreamReader(
					client.getInputStream()));
			mOutput = new BufferedOutputStream(client.getOutputStream());
			mClientSocket = client;
		}

		public void run() {
			try {
				log.i(tag, "============              ============");
				log.i(tag, "============              ============");
				log.i(tag, "======================================");
				log.i(tag, "ThreadHashCode:" + Thread.class.hashCode());
				if (!isInitProxyInfo) {
					log.i(tag, "========get proxy target=====");
					if (getRTSPServerInfo(mInput)) {
						isInitProxyInfo = true;
						log.i(tag,
								"client->target:"
										+ String.format("rtsp://%s:%d",
												request.mProxyUrl.host,
												request.mProxyUrl.getPort())
										+ "->"
										+ String.format("rtsp://%s:%d",
												request.host, request.port));
					} else {
						log.e(tag, "========proxy fail=====");
						Response.returnErr(mOutput, Response.STATUS_BAD_REQUEST);
					}
				}
				int readSize = 0;
				while (!isClosedSocket) {
					log.e(tag, "========proxy request=====");
					readSize = proxyRequest(mInput, s_mInput, s_mOutput);
				}
				mOutput.close();
				log.i(tag, "complete:" + readSize);
			} catch (IllegalStateException e) {
				e.printStackTrace();
				Response.returnErr(mOutput, Response.STATUS_Method_Not_Allowed);
				return;
			} catch (MalformedURLException e) {
				e.printStackTrace();
				Response.returnErr(mOutput,
						Response.STATUS_INTERNAL_SERVER_ERROR);
			} catch (IOException e) {
				e.printStackTrace();
				Response.returnErr(mOutput,
						Response.STATUS_INTERNAL_SERVER_ERROR);
			}finally{
				stopProxy();
			}
		}

		private void stopProxy() {
			interrupt();
			try {
				mInput.close();
				mOutput.close();
				mClientSocket.close();
				s_mInput.close();
				s_mOutput.close();
				targetSocket.close();
				connectingSocketSet.remove(targetSocket);
			} catch (Exception e) {
				e.printStackTrace();
			}
			log.e(this, "close proxy:" + request.mProxyUrl);
			log.e(this, "close client:" + request.url);
			log.e(this, "");
			log.e(this, "");
			log.e(this, "###########################################");
			log.e(this, "");
			log.e(this, "");

		}

		private boolean getRTSPServerInfo(BufferedReader mInput2)
				throws MalformedURLException, IOException {
			// int len = -1;
			// int readBytes = 0;
			String firstLine;
			if ((firstLine = mInput2.readLine()) == null)
				throw new SocketException("Client disconnected");
			Request request = Request.parseRequestFirst(firstLine);
			// Socket s = new Socket(new Proxy(Proxy.Type.SOCKS,
			// new InetSocketAddress("socks.mydom.com", 1080)));
			if (request.mProxyUrl == null) {
				Response.returnErr(mOutput, "request.mProxyUrl not is null !"
						+ Response.STATUS_BAD_REQUEST);
				return false;
			}
			this.request = request;
			//////////////////////////////
//			targetSocket = new Socket(request.mProxyUrl.host,
//					request.mProxyUrl.getPort());
//			log.i(tag, "create socket:" + request.mProxyUrl.host + ":"
//					+ request.mProxyUrl.getPort());
//			targetSocket.setSoTimeout(TIMEOUT_SOCKET);
			//////////////////////////////
			targetSocket =  mCameraProxyConnectPool.getSocketMap().get(request.gateway).getSocket();
			if (isStreaming(targetSocket)) {
				Response.returnErr(mOutput, "The camera be connect by other client !"
						+ Response.STATUS_Method_Not_Allowed);
				return false ;
			}
			connectingSocketSet.add(targetSocket);
			//TODO 使用长连接服务中的socket替代主动连接请求
			//////////////////////////////
			s_mInput = new BufferedInputStream(targetSocket.getInputStream());
			s_mOutput = new BufferedWriter(new OutputStreamWriter(
					targetSocket.getOutputStream()));
			// proxy data
			String str;
			String dstUrl = request.method + " " + request.mProxyUrl.toString()
					+ Request.version;
			log.i(tag, "send:" + dstUrl);
			s_mOutput.write(dstUrl);

			while ((str = mInput2.readLine()) != null) {
				log.i(tag, "send:" + str + "<<<");
				if (str.length() == 0) {
					log.i(tag, "send:" + "CRLF:" + CRLF);
					s_mOutput.write(str + CRLF);
					s_mOutput.flush();
					break;
				} else {
					s_mOutput.write(str + CRLF);
				}

			}
			byte[] responseBuffer = new byte[BUFFER_SIZE];
			int readSize = 0;
			log.i(tag, "proxy response");
			while ((readSize = s_mInput.read(responseBuffer)) != -1) {
				String response = new String(responseBuffer, 0, readSize);
				log.i(tag, "response:" + readSize + ":" + response);
				response = replaceProxyIPPort(response);
				mOutput.write(response.getBytes());
				mOutput.flush();
				break;
			}
			return true;
		}

		/**
		 * proxy read the client data and send to target<br>
		 * sent date that the target response to client ;
		 * 
		 * @param mInput2
		 * @param s_mInput
		 * @param s_mOutput
		 * @return
		 * @throws IOException
		 */
		private int proxyRequest(BufferedReader mInput2,
				BufferedInputStream s_mInput, BufferedWriter s_mOutput)
				throws IOException {
			StringBuilder sb = new StringBuilder();
			String str;
			boolean isSteaming = false;
			int readSize;
			String contentlength_old = "";
			int contentlength_old_size, contentlength_new_size = 0, contentlength_real_size = 0;
			String contentlength_new = "";
			boolean isContentChange = false;
			boolean isCalRealSize = false;
			String KEY_CONTENTLENGTH = "Content-Length: ";
			log.i(tag, "=======read client request====");
			while ((str = mInput2.readLine()) != null) {
				if (str.length() == 0) {
					log.i(tag, "target return string:" + str);
					// System.out.println("request.url" + request.url);
					// System.out.println("mProxyUrl.url" +
					// request.mProxyUrl.url);
					str = sb.toString().replace(request.url,
							request.mProxyUrl.url); // 直接替换
					// str = replaceRequestIPPort(sb.toString());
					log.i(tag, "send to target:" + str);
					if (str.startsWith(Request.Method_PLAY)) {
						// start proxy RTP data TCP steam
						isSteaming = true;
					}
					if (str.startsWith(Request.Method_DESCRIBE)) {
						// replace ipINFO case content length change
						isContentChange = true;
					} else if (str.startsWith("RTSP/1.0")) {
						Response.returnErr(mOutput,
								Response.STATUS_Method_Not_Allowed);
						return 0;
					}
					s_mOutput.write(str + CRLF);
					s_mOutput.flush();
					break;
				} else {
					sb.append(str);
					sb.append(CRLF);
				}

			}
			byte[] responseBuffer;
			responseBuffer = new byte[BUFFER_SIZE];
			log.i(tag, "========read the target response========");
			boolean isreplaceRtpInfo = false;
			while ((readSize = s_mInput.read(responseBuffer)) != -1) {

				if (isSteaming) {
					// if (!isreplaceRtpInfo) {
					// log.i(tag, "send steaming data to client,readSize:" +
					// readSize+"<<<");
					// String response = new String(responseBuffer, 0,
					// readSize);
					// response = replaceProxyIPPort(response);
					// log.i(tag, response);
					// mOutput.write(response.getBytes("UTF-8"));
					// mOutput.flush();
					// isreplaceRtpInfo =true;
					// }else{
					mOutput.write(responseBuffer, 0, readSize);
					mOutput.flush();
					// }

				} else {
					String response = new String(responseBuffer, 0, readSize);

					if (isContentChange) {
						log.i(tag, "<<<<<<<<<<<<<<<<<<<<<<<<<<");
						log.i(tag, "send to client:" + response + "<<<");
						String[] responseline = response.split(CRLF);
						for (String line : responseline) {
							if (isCalRealSize) {
								contentlength_real_size += line
										.getBytes(charset).length
										+ CRLF.length();
							}
							if (line.startsWith(KEY_CONTENTLENGTH)) {
								contentlength_old = line;
								contentlength_old_size = Integer
										.parseInt(contentlength_old.split(" ")[1]);
								isCalRealSize = true;
								log.i(tag, "sdp contentlength_old:"
										+ contentlength_old + "<<<");
							}

						}
						contentlength_real_size -= CRLF.length();// FIXME BUG 这里为什么要‘-’一下 是猜的
						isCalRealSize = false;
						log.i(tag, "sdp length:" + contentlength_real_size
								+ "<<<");
					}

					response = replaceProxyIPPort(response);
					// recalu conentlength
					if (isContentChange) {
						String[] responseline = response.split(CRLF);
						for (String line : responseline) {
							if (isCalRealSize) {
								contentlength_new_size += line
										.getBytes(charset).length
										+ CRLF.length();
							}
							if (line.startsWith(KEY_CONTENTLENGTH)) {
								isCalRealSize = true;
							}
						}
						contentlength_new_size -= CRLF.length();// FIXME BUG why should "-="
						contentlength_new = KEY_CONTENTLENGTH
								+ contentlength_new_size;

						log.i(tag, "sdp contentlength_new:" + contentlength_new
								+ "<<<");
					}
					if (isContentChange) {
						log.i(tag, "sdp contentlength_old:" + contentlength_old
								+ "<<<");
						response = response.replace(contentlength_old,
								contentlength_new);
						isContentChange = false;
					}
					log.i(tag, "send to client:" + response + "<<<");
					mOutput.write(response.getBytes(charset));
					mOutput.flush();
					break;
				}

			}
			return readSize;
		}

		final static String Method_PLAY = "PLAY";

		/**
		 * replace target IP to Proxy server IP
		 * 
		 * @param str
		 * @return
		 */
		private String replaceProxyIPPort(String str) {
			log.i(tag, "---------replaceProxyIPPort---------");

			str = str.replace(String.format("rtsp://%s:%d",
					request.mProxyUrl.host, request.mProxyUrl.getPort()),
					String.format("rtsp://%s:%d", request.host, request.port));
			return str;
		}

		private String replaceRequestIPPort(String str) {
			log.i(tag, "---------replace request---------");
			log.i(tag, str);
			str = str.replace(String.format("/?url=rtsp://%s:%d/",
					request.mProxyUrl.host, request.mProxyUrl.getPort()), "");
			log.i(tag,
					String.format("/?url=rtsp://%s:%d/",
							request.mProxyUrl.host, request.mProxyUrl.getPort())
							+ "->" + "");
			str = str.replace(String.format("rtsp://%s:%d", request.host,
					request.port), String.format("rtsp://%s:%d",
					request.mProxyUrl.host, request.mProxyUrl.getPort()));
			log.i(tag, ">>" + str);
			return str;
		}

		/**
		 * 
		 * @param mInput2
		 * @param s_mInput
		 * @param s_mOutput
		 * @return
		 * @throws IOException
		 */
		private int proxy(BufferedReader mInput2, BufferedInputStream s_mInput,
				BufferedWriter s_mOutput) throws IOException {
			// String str;
			byte[] responseBuffer;
			int readSize;
			// log.i(tag,"read");
			// while ((str = mInput2.readLine()) != null) {
			//
			// str = str.replace(String.format("/?url=rtsp://%s:%d/",
			// request.mProxyUrl.host, request.mProxyUrl.getPort()),
			// "");
			// str = str.replace(String.format("rtsp://%s:%d", request.host,
			// request.port), String.format("rtsp://%s:%d",
			// request.mProxyUrl.host, request.mProxyUrl.getPort()));
			// log.i(tag,"send:" + str + "<<<");
			// if (str.length() == 0) {
			// log.i(tag,"send:" + "CRLF:" + CRLF);
			// s_mOutput.write(str + CRLF);
			// s_mOutput.flush();
			// break;
			// } else {
			// s_mOutput.write(str + CRLF);
			// }
			//
			// }
			responseBuffer = new byte[BUFFER_SIZE];
			log.i(tag, "proxy media data");
			while ((readSize = s_mInput.read(responseBuffer)) != -1) {
				mOutput.write(responseBuffer);
				mOutput.flush();
				log.i(tag, "over");
				break;
			}
			return readSize;
		}
	}

	public static void main(String[] args) throws IOException {
		RTSPProxy t = new RTSPProxy();
		t.start();
	}
	private HashSet<Socket> connectingSocketSet  = new HashSet<Socket>();
	public boolean isStreaming(Socket targetSocket) {
		return connectingSocketSet.contains(targetSocket);
	}

	public static String tag = "rtspproxy";

	public static class log {
		public static void i(Object tag, Object log) {
			System.out.println(log);
		}

		public static void e(Object tag, Object log) {
			System.out.println(log);
		}
	}
}
